package com.googlecode.objectify.hbase;

import com.google.appengine.api.NamespaceManager;
import com.google.appengine.api.datastore.Cursor;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.FetchOptions;
import com.google.appengine.api.datastore.Index;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.KeyRange;
import com.google.appengine.api.datastore.PreparedQuery;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.QueryResultIterator;
import com.google.appengine.api.datastore.Transaction;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author Secondary User
 */
public class HBaseScanPreparedQuery implements PreparedQuery
{

	private final static char AFTER_SPLIT_CHAR = Key.SPLIT_CHAR + 1;
	private int chunkSize = 30;
	private final Future<ResultScanner> scan;
	private /*final*/ Future<Result[]> firstResults; //Should only be written once.

	public HBaseScanPreparedQuery(Transaction txn, Query query, final HTableInterface main, HTableInterface index, final ExecutorService exec)
	{
		if (!query.needsIndex())
		{
			String prefixTemp;
			Key parent = query.getAncestor();
			if (parent != null)
			{
				prefixTemp = NamespaceManager.get() + Key.SPLIT_CHAR + "P" + Key.SPLIT_CHAR + parent.toKeyString() + Key.SPLIT_CHAR;
			}
			else
			{
				prefixTemp = NamespaceManager.get() + Key.SPLIT_CHAR + "K" + Key.SPLIT_CHAR;
			}
			String kind = query.getKind();
			if (kind != null && !kind.isEmpty())
			{
				prefixTemp += kind;
			}
			final String prefix = prefixTemp;
			scan = exec.submit(new Callable<ResultScanner>()
			{
				@Override
				public ResultScanner call() throws Exception
				{
					ResultScanner toReturn = main.getScanner(new Scan(Bytes.toBytes(prefix + Key.SPLIT_CHAR), Bytes.toBytes(prefix + AFTER_SPLIT_CHAR)));
					firstResults = getResultsAsync(toReturn, chunkSize, exec);
					return toReturn;
				}
			});
		}
		else
		{
			scan = null; //TODO: Implement scanner for indexes.
		}
	}

	private Future<Result[]> getResultsAsync(final ResultScanner scanner, final int chunkSize, ExecutorService exec)
	{
		return exec.submit(new Callable<Result[]>()
		{
			@Override
			public Result[] call() throws Exception
			{
				return scanner.next(chunkSize);
			}
		});
	}

	@Override
	public int countEntities(FetchOptions fetchOpts)
	{
		try
		{
			ResultScanner scn = scan.get();
			Result[] results = firstResults.get();
			if (results == null)
				return 0;
			int count = 0;
			while (results != null && results.length > 0)
			{
				try
				{
					count += results.length;
					results = scn.next(fetchOpts.getChunkSize());
				}
				catch (IOException ex)
				{
					throw new RuntimeException(ex);
				}
			}
			return count;
		}
		catch (InterruptedException ex)
		{
			throw new RuntimeException(ex);
		}
		catch (ExecutionException ex)
		{
			throw new RuntimeException(ex);
		}
	}

	@Override
	public QueryResultIterator<Entity> asQueryResultIterator(final FetchOptions fetchOpts)
	{
		final Result[] firstResultsGot;
		final ResultScanner scn;
		final Result next;
		
		try
		{
			firstResultsGot = firstResults.get();
			scn = scan.get();
		}
		catch (InterruptedException ex)
		{
			throw new RuntimeException(ex);
		}
		catch (ExecutionException ex)
		{
			throw new RuntimeException(ex);
		}
		return new QueryResultIterator<Entity>()
		{
			Result[] results = firstResultsGot;
			int pos = 0;
			
			@Override
			public Cursor getCursor()
			{
				throw new UnsupportedOperationException("Not supported yet.");
			}

			@Override
			public List<Index> getIndexList()
			{
				throw new UnsupportedOperationException("Not supported yet.");
			}

			@Override
			public boolean hasNext()
			{
				if (pos == results.length)
					return false;
				return true;
			}

			@Override
			public Entity next()
			{
				if (pos == results.length)
					throw new IllegalStateException("TODO:");
				{
					pos = 0;
					try
					{
						results = scn.next(fetchOpts.getChunkSize());
					}
					catch (IOException ex)
					{
						throw new RuntimeException(ex);
					}
				}
                                throw new UnsupportedOperationException("Not supported yet.");
			}

			@Override
			public void remove()
			{
				throw new UnsupportedOperationException("Not supported yet.");
			}
		};
	}
}
